package com.lubase.wfengine.consumer.memory;

import com.alibaba.fastjson.JSON;
import com.lubase.orm.extend.ExtendAppLoadCompleteService;
import com.lubase.wfengine.auto.entity.WfCallbackEntity;
import com.lubase.wfengine.auto.entity.WfTEventEntity;
import com.lubase.wfengine.consumer.ProcessEngineEventService;
import com.lubase.wfengine.consumer.UpdateBisTableService;
import com.lubase.wfengine.remote.impl.WFEngineMemoryMQService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;

import java.util.concurrent.ConcurrentLinkedQueue;

@Slf4j
@Service
@ConditionalOnProperty(name = "lubase.wf-engine.mq-type", havingValue = "memory")
public class MemoryMQConsumerService implements ExtendAppLoadCompleteService {
    @Autowired
    WFEngineMemoryMQService wfEngineMemoryMQService;

    @Autowired
    ProcessEngineEventService processEngineEventService;
    @Autowired
    UpdateBisTableService updateBisTableService;

    @Override
    public void LoadCompleted(ApplicationContext applicationContext) {
        // 启动线程 监听队列
        new Thread(() -> {
            while (true) {
                try {
                    ConcurrentLinkedQueue<String> queue = wfEngineMemoryMQService.getEngineQueue();
                    if (!queue.isEmpty()) {
                        String message = queue.poll();
                        WfTEventEntity entity = null;
                        try {
                            entity = JSON.parseObject(message, WfTEventEntity.class);
                            processEngineEventService.processWftEvent(entity);
                        } catch (Exception exception) {
                            log.error("processWftEvent 执行失败，event id is{}", message, exception);
                        }
                    }
                } catch (Exception e) {
                    log.error("MemoryMQConsumerService error", e);
                }
            }
        }).start();

        // 启动线程 监听队列
        new Thread(() -> {
            while (true) {
                try {
                    ConcurrentLinkedQueue<String> queue = wfEngineMemoryMQService.getUpdateTableQueue();
                    if (!queue.isEmpty()) {
                        String message = queue.poll();

                        WfCallbackEntity entity = null;
                        try {
                            entity = JSON.parseObject(message, WfCallbackEntity.class);
                            updateBisTableService.updateBisTable(entity);
                        } catch (Exception exception) {
                            log.error("processWftEvent 执行失败，event id is{}", message, exception);
                        }
                    }
                } catch (Exception e) {
                    log.error("MemoryMQConsumerService error", e);
                }
            }
        }).start();
    }

    @Override
    public void clearData() {

    }
}
